package com.wjk.kylin.rpc.core.registry.impl;

import com.wjk.kylin.rpc.core.registry.Register;
import com.wjk.kylin.rpc.core.util.RpcException;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * CreateMode.PERSISTENT ：持久节点，一旦创建就保存到硬盘上面
 * CreateMode.SEQUENTIAL ： 顺序持久节点会在 path后面，自动添加一个递增的10位整数，
 * CreateMode.EPHEMERAL ：临时节点，创建以后如果断开连接则该节点自动删除
 * CreateMode.EPHEMERAL_SEQUENTIAL ：顺序临时节点，会在 path后面，自动添加一个递增的10位整数
 *
 * @author wjk
 */
public class ZookeeperRegister implements Register {
    protected static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperRegister.class);

    /**
     * 服务地址列表
     */
    private static final String ZK_REGISTRY_PATH = "/kylin/rpc";
    private ZkClient client;

    private volatile ConcurrentMap<String, TreeSet<String>> discoveryData = new ConcurrentHashMap<>();
    private final Map<String, TreeSet<String>> registryData = new HashMap<>();

    public ZookeeperRegister() {
    }

    /**
     * 连接zookeeper
     *
     * @param registryAddress
     * @return
     */
    private ZkClient connectServer(String registryAddress) {
        return new ZkClient(registryAddress, 30000, 30000);
    }

    //==============================create==============================================

    /**
     * 创建根目录
     *
     * @param client
     */
    private void createRootNode(ZkClient client) {
        boolean exists = client.exists(ZK_REGISTRY_PATH);
        if (!exists) {
            client.createPersistent(ZK_REGISTRY_PATH, true);
        }
    }

    /**
     * 创建持久顺序节点
     *
     * @param servicePath
     */
    private void createNodePersistent(String servicePath) {
        if (StringUtils.isBlank(servicePath)) {
            return;
        }
        boolean exists = client.exists(ZK_REGISTRY_PATH + "/" + servicePath);
        if (!exists) {
            client.createPersistent(ZK_REGISTRY_PATH + "/" + servicePath);
            LOGGER.debug("kylin-rpc provider zk create persistent node : ({}/{})", ZK_REGISTRY_PATH, servicePath);
        } else {
            LOGGER.debug("kylin-rpc provider zk create persistent node is exists : ({}/{})", ZK_REGISTRY_PATH, servicePath);
        }
    }

    /**
     * 创建临时顺序子节点
     * 重启服务，可能节点还存在，没有自动删除，也会不在创建，等会自动删除了，就会找不到该节点
     *
     * @param servicePath
     * @param addressPath
     */
    private void createNodeEphemeral(String servicePath, String addressPath) {
        if (StringUtils.isBlank(servicePath) || StringUtils.isBlank(addressPath)) {
            return;
        }
        boolean exists = client.exists(ZK_REGISTRY_PATH + "/" + servicePath + "/" + addressPath);
        if (!exists) {
            client.createEphemeral(ZK_REGISTRY_PATH + "/" + servicePath + "/" + addressPath);
            LOGGER.debug("kylin-rpc provider zk create ephemeral node : ({}/{}/{})", ZK_REGISTRY_PATH, servicePath, addressPath);
        } else {
            LOGGER.debug("kylin-rpc provider zk create ephemeral node is exists : ({}/{}/{})", ZK_REGISTRY_PATH, servicePath, addressPath);
        }
    }

    private List<String> getRootNodeChildren() {
        return client.getChildren(ZK_REGISTRY_PATH);
    }

    private List<String> getNodeChildren(String servicePath) {
        if (StringUtils.isBlank(servicePath)) {
            return null;
        }
        return client.getChildren(ZK_REGISTRY_PATH + "/" + servicePath);
    }

    private void deleteNodeChildren(String servicePath, String addressPath) {
        if (StringUtils.isBlank(servicePath) || StringUtils.isBlank(addressPath)) {
            return;
        }
        boolean delete = client.delete(ZK_REGISTRY_PATH + "/" + servicePath + "/" + addressPath);
        LOGGER.debug("kylin-rpc provider zk delete node : ({}/{}/{}),flag:{}", ZK_REGISTRY_PATH, servicePath, addressPath, delete);
    }


    /**
     * 客户端监听
     *
     * @param servicePath
     */
    public void discoverySubscribeChildChanges(String servicePath) {
        client.subscribeChildChanges(ZK_REGISTRY_PATH + "/" + servicePath, new ZkChildListenerDiscoveryServicePath());
    }

    /**
     * 服务端监听
     *
     * @param servicePath
     */
    public void registrySubscribeChildChanges(String servicePath) {
        client.subscribeChildChanges(ZK_REGISTRY_PATH + "/" + servicePath, new ZkChildListenerRegistryServicePath());
    }

    //================================对外提供接口====================================

    /**
     * 启动服务 连接注册中心
     *
     * @param registryAddress
     */
    @Override
    public void start(String registryAddress) {
        if (StringUtils.isBlank(registryAddress)) {
            throw new RpcException("kylin-rpc init zookeeper address is empty");
        }
        try {
            client = connectServer(registryAddress);
        } catch (Exception e) {
            LOGGER.error(">>>>>>>>>>> kylin-rpc connect zookeeper client error address = {},e:", registryAddress, e);
            throw new RpcException(e);
        }
        if (client != null) {
            //创建根目录
            createRootNode(client);
        } else {
            throw new RpcException("kylin-rpc init zookeeper client is empty");
        }

    }

    /**
     * 停止服务
     */
    @Override
    public void stop() {
        registryData.clear();
        discoveryData.clear();
    }

    /**
     * 服务端注册
     *
     * @param keys  service key
     * @param value service value/ip:port
     * @return
     */
    @Override
    public boolean registry(Set<String> keys, String value) {
        if (CollectionUtils.isEmpty(keys) || StringUtils.isBlank(value)) {
            return false;
        }
        for (String key : keys) {
            //加入缓存
            TreeSet<String> values = registryData.computeIfAbsent(key, k -> new TreeSet<>());
            values.add(value);

            //创建接口节点
            createNodePersistent(key);
            //创建接口提供者地址接口
            createNodeEphemeral(key, value);

            //加入监听
            registrySubscribeChildChanges(key);
        }
        return true;
    }

    /**
     * 服务关闭
     *
     * @param keys
     * @param value
     * @return
     */
    @Override
    public boolean remove(Set<String> keys, String value) {
        if (CollectionUtils.isEmpty(keys) || StringUtils.isBlank(value)) {
            return false;
        }
        for (String key : keys) {
            //删除 接口提供者地址接口
            deleteNodeChildren(key, value);
        }
        return true;
    }

    /**
     * 客户端 启动
     *
     * @param keys
     */
    @Override
    public void discovery(Set<String> keys) {
        if (CollectionUtils.isEmpty(keys)) {
            return;
        }
        for (String key : keys) {
            //加入缓存
            discovery(key);
            //节点加入监听
            discoverySubscribeChildChanges(key);
        }
    }

    /**
     * 发现服务
     *
     * @param key service key
     * @return
     */
    @Override
    public TreeSet<String> discovery(String key) {
        if (StringUtils.isBlank(key)) {
            return null;
        }
        TreeSet<String> treeSet = discoveryData.get(key);
        if (!CollectionUtils.isEmpty(treeSet)) {
            return treeSet;
        }

        List<String> children = getNodeChildren(key);
        if (CollectionUtils.isEmpty(children)) {
            return null;
        }

        treeSet = new TreeSet<>(children);
        discoveryData.put(key, treeSet);

        return treeSet;
    }

    /**
     *
     */
    class ZkChildListenerDiscoveryServicePath implements IZkChildListener {
        @Override
        public void handleChildChange(String path, List<String> list) throws Exception {
            LOGGER.debug("kylin-rpc consumer zk listener child change node : {}", path);
            LOGGER.debug("kylin-rpc consumer zk listener child change node child has: {}", list.toString());

            if (StringUtils.isNotBlank(path) && path.startsWith(ZK_REGISTRY_PATH)) {
                String servicePath = path.replace(ZK_REGISTRY_PATH + "/", "");
                if (StringUtils.isNotBlank(servicePath)) {
                    discoveryData.put(servicePath, CollectionUtils.isEmpty(list) ? new TreeSet<>() : new TreeSet<>(list));
                    //再次加入监听
//                    discoverySubscribeChildChanges(servicePath);
                }
            }
        }
    }

    /**
     *
     */
    class ZkChildListenerRegistryServicePath implements IZkChildListener {
        @Override
        public void handleChildChange(String path, List<String> list) throws Exception {
            LOGGER.debug("kylin-rpc provider zk listener child change node : {}", path);
            LOGGER.debug("kylin-rpc provider zk listener child change node child has: {}", list.toString());

            if (StringUtils.isNotBlank(path) && path.startsWith(ZK_REGISTRY_PATH)) {
                String servicePath = path.replace(ZK_REGISTRY_PATH + "/", "");
                if (StringUtils.isNotBlank(servicePath)) {
                    TreeSet<String> treeSet = registryData.get(servicePath);
                    if (!CollectionUtils.isEmpty(treeSet)) {
                        TreeSet<String> addressPathSet = new TreeSet<>(treeSet);
                        if (!CollectionUtils.isEmpty(list)) {
                            treeSet.addAll(list);
                        }
                        for (String addressPath : addressPathSet) {
                            //缓存存在，zk不存在，则创建
                            if (treeSet.contains(addressPath) && !list.contains(addressPath)) {
                                createNodeEphemeral(servicePath, addressPath);
                            } else if (!treeSet.contains(addressPath) && list.contains(addressPath)) {
                                //缓存不存在，zk存在，则删除
                                deleteNodeChildren(servicePath, addressPath);
                            }
                        }
                    } else if (!CollectionUtils.isEmpty(list)) {
                        for (String addressPath : list) {
                            //缓存不存在，zk存在，则删除
                            deleteNodeChildren(servicePath, addressPath);
                        }
                    }

                    //再次加入监听 会造成多个监听
//                    registrySubscribeChildChanges(servicePath);
                }
            }
        }
    }
}
